package com.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.base.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/*
 * @description: TODO
 * @param
 * @return
 * @author liuyun
 * @date 2022-3-18 22:12
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private Connection conn;
    private ValueState<Boolean> tableCreatedState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取phoenix连接对象  一个并行度一个连接对象
        conn = JdbcUtil.getJdbcConnect(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL, null, null);
        tableCreatedState = getRuntimeContext().getState(
                new ValueStateDescriptor<Boolean>("tableCreatedState", Boolean.class));

    }

    @Override
    public void close() throws Exception {
        // 关闭phoenix连接对象
        if (conn != null && !conn.isClosed()) {
            conn.close();
        }
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context context) throws Exception {

        // 1. 建表
        checkTable(value);

        // 2. 把这条数据写入到对应的表中

        writeToPhoenix(value);

    }

    private void writeToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {

        JSONObject data = value.f0;
        //System.out.println(data);

        TableProcess tp = value.f1;

        StringBuilder sql = new StringBuilder();

        sql
                .append("upsert into ")
                .append(tp.getSink_table())
                .append("(")
                //拼接字段名
                .append(tp.getSink_columns())
                .append(") values(")
                //拼接占位符 有几个字段, 就拼接几个问号
                .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
                .append(")");

        System.out.println("插入语句" + sql);

        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 使用数据中的每个字段的值, 给占位符赋值

        String[] columns = tp.getSink_columns().split(",");

        for (int i = 0; i < columns.length; i++) {
            String key = columns[i];
            Object v = data.get(key);
            ps.setString(i + 1, v == null ? null : v.toString());

        }


        ps.execute();

        conn.commit();

        ps.close();
    }


    /*
     * @description: TODO
     * @param
     * @return
     * @author liuyun
     * @date 2022-3-18 22:54
     */
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {

        if (tableCreatedState.value() == null) {
            TableProcess tp = value.f1;

            // 建表 执行sql ddl 建表语句
            // create table if not exists t(id varchar, age  varchar, name varchar , constraint pk primary key(id, age))SALT_BUCKETS = 4;
            // phoenix中的表的表必须有主键!!!
            StringBuilder sql = new StringBuilder();
            // 拼接sql语句  ToDo
            sql
                    .append("create table if not exists ")
                    .append(tp.getSink_table())
                    .append("(")
                    .append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                    .append(", constraint pk primary key(")
                    .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                    .append("))")
                    .append(tp.getSink_extend() == null ? "" : tp.getSink_extend()); // "null"

            System.out.println("建表语句: " + sql);
            PreparedStatement ps = conn.prepareStatement(sql.toString());

            // 一般情况下, sql语句中可能会存在占位符 ? , 需要给占位符进行赋值
            ps.execute();
            conn.commit();
            ps.close();

            tableCreatedState.update(true);
        }


    }
}